package com.wjw.redis.stream.handler;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.stream.ByteRecord;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.PendingMessage;
import org.springframework.data.redis.connection.stream.PendingMessages;
import org.springframework.data.redis.connection.stream.PendingMessagesSummary;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.List;

/**
 * @author wjwei
 */
@Slf4j
public class PendingQueueHandle {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    // 假设这是要进行特殊处理的ID，现实情况下应该通过配置获
    private static final String SPECIAL_ID = "1706233940011-0";

    /**
     * 处理挂起消息
     *
     * @param key   streamKey
     * @param group 消费者组
     */
    public void handlePendingMessages(String key, String group) {
        if (!StringUtils.hasText(key) || !StringUtils.hasText(group)) {
            log.warn("无效的参数: key 或 group 为空");
            return;
        }
        try {
            //判断是否存在挂起的消息 XPENDING stream group [start stop count] [consumer]
            PendingMessagesSummary pending = redisTemplate.opsForStream().pending(key, group);
            long size;
            if (pending == null || (size = pending.getTotalPendingMessages()) <= 0) {
                log.debug("消息队列 [{}:{}] 暂无挂起消息", key, group);
                return;
            }

            log.info("消息队列 [{}:{}] 挂起消息数量: {}", key, group, size);
            handlePendingMessagesInRange(key, group, pending);
        } catch (Exception e) {
            log.error("处理挂起消息时发生异常", e);
        }
    }

    private void handlePendingMessagesInRange(String key, String group, PendingMessagesSummary pending) {
        String minId = pending.minMessageId(), maxId = pending.maxMessageId();
        //Xpending stream group [start stop count] [consumer]
        PendingMessages msgIds = redisTemplate.opsForStream().pending(key, group, Range.closed(minId, maxId), pending.getTotalPendingMessages());
        log.info("消息队列 [{}:{}] 挂起消息 ID: {}", key, group, msgIds);

        for (PendingMessage msgId : msgIds) {
            String id = msgId.getId().getValue();
            log.info("消息队列 [{}:{}] 处理挂起消息 ID: {}", key, group, id);
            //XRANGE key start end [COUNT count]
            List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().range(key, Range.just(msgId.getIdAsString()));
            if (records != null && !records.isEmpty()) {
                // 开始 结束 id相同，只返回 一个消息结果
                MapRecord<String, Object, Object> record = records.get(0);
                if (SPECIAL_ID.equals(id)) {
                    processSpecialMessage(key, group, record);
                } else {
                    log.info("消息队列 [{}:{}] 普通消息: {}", key, group, record);
                }
            }
        }
        log.info("消息队列 [{}:{}] 处理挂起消息结束", key, group);
    }

    private void processSpecialMessage(String key, String group, MapRecord<String, Object, Object> record) {
        MapRecord<String, Object, Object> copy = record.withId(RecordId.autoGenerate());
        log.info("消息队列 [{}:{}] 处理特殊消息, 新ID: {}", key, group, copy);
        //XADD stream-name id field value [field value]
        RecordId add = redisTemplate.opsForStream().add(copy);
        log.info("消息队列 [{}:{}] 消息添加结果: {}", key, group, add);
        log.info("消息队列 [{}:{}] 确认处理旧消息: {}", key, group, record);
        //XACK stream group id [id id ...]
        Long siz = redisTemplate.opsForStream().acknowledge(group, record);
        log.info("消息队列 [{}:{}] 确认处理结果: {}", key, group, siz);
        log.info("消息队列 [{}:{}] 删除旧消息: {}", key, group, record);
        //XDEL key ID [ID ...]
        siz = redisTemplate.opsForStream().delete(record);
        log.info("消息队列 [{}:{}] 删除结果: {}", key, group, siz);
    }

    /**
     * 转移消息
     * XCLAIM stream group new-consumer min-idle-time id [ID ...]
     *
     * @param key         streamKey
     * @param group       旧消费者组
     * @param newConsumer 新消费者
     * @param minIdleTime 最小空闲时间
     * @param recordId    消息id
     */
    public void xClaim(String key, String group, String newConsumer, long minIdleTime, String recordId) {
        if (key == null || group == null || newConsumer == null || recordId == null || minIdleTime < 0) {
            log.info("Invalid input parameters");
            return;
        }

        try {
            // 使用类型断言以确保转换安全
            @SuppressWarnings("unchecked")
            List<ByteRecord> retVal = (List<ByteRecord>) redisTemplate.execute((RedisCallback<List<ByteRecord>>) connection -> connection.streamCommands().xClaim(key.getBytes(), group, newConsumer, Duration.ofSeconds(minIdleTime), RecordId.of(recordId)));

            if (retVal != null) {
                for (ByteRecord byteRecord : retVal) {
                    log.info("改了消息的消费者：id={}, value={}", byteRecord.getId(), byteRecord.getValue());
                }
            }
        } catch (Exception e) {
            log.error("Error claiming records in Redis Stream", e);
        }
    }

}
